package cn.lyf.myblog.dataFlow;

import java.util.*;
import java.util.function.*;
import java.util.stream.*;
import java.util.concurrent.*;

public class Flow<T> {
    final private Map<String, UnaryOperator<T>> nodes = new ConcurrentHashMap<>();
    final private Map<String, List<String>> edges = new ConcurrentHashMap<>();
    final private Map<String, Flow<T>> map = new ConcurrentHashMap<>();
    final private Set<String> jump = new CopyOnWriteArraySet<>();
    // 并发等级
    final private int level;

    public static <E> Flow<E> of(Class<E> e) {
        return new Flow<E>();
    }

    public Flow() {
        this.level = 0;
    }

    public Flow(int level) {
        this.level = level;
    }

    public FlowUtils<T> utils() {
        return new FlowUtils<>(this);
    }

    public StreamUtils<T> stream() {
        return new StreamUtils<>(this);
    }

    // 设置处理节点
    public Flow<T> setHandler(String k, UnaryOperator<T> fx) {
        nodes.put(k, fx);
        return this;
    }

    // 移除处理节点
    public Flow<T> removeHandler(String k) {
        nodes.remove(k);
        return this;
    }

    // 短路处理节点
    public Flow<T> shortcut(String k) {
        if (jump.add(k))
            return this;
        else
            return null;
    }

    // 恢复处理节点
    public Flow<T> unshortcut(String k) {
        if (jump.remove(k))
            return this;
        else
            return null;
    }

    // 从K节点开始处理data
    public void start(String k, T data) {
        start(k, data, level);
    }

    // 用子流替换k节点
    public Flow<T> map(String key, Flow<T> subFlow, UnaryOperator<T> fx) {
        map(key, subFlow);
        return subFlow.setHandler(key, fx);
    }

    // 用子流细化k节点
    public Flow<T> map(String key, Flow<T> subFlow) {
        map.put(key, subFlow);
        UnaryOperator<T> fx = nodes.get(key);
        if (fx != null) {
            subFlow.setHandler(key, fx);
            for (String to : edges.getOrDefault(key, new ArrayList<>())) {
                subFlow.link(key, to);
                subFlow.map.put(to, this);
            }
        }
        return this;
    }

    // 移除子流
    public Flow<T> removemap(String key) {
        map.remove(key);
        return this;
    }

    // 链接from和to节点
    public Flow<T> link(String from, String to) {
        List<String> l = edges.get(from);
        if (l == null) {
            l = new CopyOnWriteArrayList<>();
            edges.put(from, l);
        }
        if (l.add(to))
            return this;
        return null;
    }

    // 断开from和to节点链接
    public Flow<T> unlink(String from, String to) {
        List<String> l = edges.get(from);
        if (l != null && l.remove(to))
            return this;
        return null;
    }

    // 以等级d从K节点开始处理data
    private void start(String k, T data, int d) {
        if (!jump.contains(k)) {
            Flow<T> subFlow = map.get(k);
            if (subFlow != null) {
                // 不会生成多余的线程
                subFlow.start(k, data, d - 1);
                return;
            }
            UnaryOperator<T> fx = nodes.get(k);
            if (fx == null)
                return;
            data = fx.apply(data);
            if (data == null)
                return;
        }
        List<String> list = edges.getOrDefault(k, new ArrayList<>());
        if (d < 0 || list.size() < 2) {
            for (String v : list) {
                // start(v, DeepCopy.copy(data), d - 1);
                start(v, data, d);
            }
        } else {
            final T msg = data;
            for (String v : list) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        start(v, DeepCopy.copy(msg), d - 1);
                    }
                }).start();
            }
        }
    }

    /*
     * 绘制Mermaid图像
     */
    public synchronized void toMermaid(String[] starts) {
        System.out.println("```mermaid");
        System.out.println("graph TD");
        System.out.println("classDef black fill:#222,color:#fff,stroke:#000,stroke-width:2px;");
        System.out.println("classDef red fill:#f22,color:#fff,stroke:#000,stroke-width:2px;");
        for (String s : starts)
            toMermaid(null, s);
        System.out.println("```");
    }

    private void toMermaid(String f, String s) {
        System.out.println(s + "(('" + s + "'))");
        System.out.println("class " + s + " " + (map.get(s) == null ? "black" : "red"));
        if (f != null)
            System.out.println(f + "-->" + s);
        List<String> ns = edges.get(s);
        if (ns == null)
            return;
        for (String next : ns)
            toMermaid(s, next);
    }

    /*
     * 简化操作的工具包
     */
    public static class FlowUtils<T> {
        private String last;
        final private Flow<T> flow;

        public FlowUtils<T> print() {
            return setHandler(i -> {
                System.out.println(i);
                return i;
            });
        }

        public FlowUtils(Flow<T> flow) {
            this.flow = flow;
        }

        public FlowUtils<T> start(String k, Consumer<T> fx) {
            return start(k, voidable(fx));
        }

        public FlowUtils<T> start(String k, Supplier<T> fx) {
            return start(k, voidable(fx));
        }

        public FlowUtils<T> start(String k, Runnable fx) {
            return start(k, voidable(fx));
        }

        public FlowUtils<T> start(String k, UnaryOperator<T> fx) {
            flow.setHandler(k, fx);
            return start(k);
        }

        public FlowUtils<T> start(String k) {
            last = k;
            return this;
        }

        public FlowUtils<T> link(String k) {
            flow.link(last, k);
            last = k;
            return this;
        }

        public FlowUtils<T> link(String k, Consumer<T> fx) {
            return link(k, voidable(fx));
        }

        public FlowUtils<T> link(String k, Supplier<T> fx) {
            return link(k, voidable(fx));
        }

        public FlowUtils<T> link(String k, Runnable fx) {
            return link(k, voidable(fx));
        }

        public FlowUtils<T> link(String k, UnaryOperator<T> fx) {
            flow.setHandler(k, fx);
            return link(k);
        }

        public FlowUtils<T> setHandler(Consumer<T> fx) {
            return setHandler(voidable(fx));
        }

        public FlowUtils<T> setHandler(Supplier<T> fx) {
            return setHandler(voidable(fx));
        }

        public FlowUtils<T> setHandler(Runnable fx) {
            return setHandler(voidable(fx));
        }

        public FlowUtils<T> setHandler(UnaryOperator<T> fx) {
            String[] ss = last.split("-");
            int n = 0;
            if (ss.length == 2)
                n = Integer.valueOf(ss[1]) + 1;
            String k = ss[0] + "-" + n;
            return link(k, fx);
        }

        public FlowUtils<T> map(Flow<T> f) {
            flow.map(last, f);
            return this;
        }

        // T->void
        private UnaryOperator<T> voidable(Consumer<T> fx) {
            return new UnaryOperator<T>() {
                public T apply(T t) {
                    fx.accept(t);
                    return t;
                };
            };
        }

        // void->T
        private UnaryOperator<T> voidable(Supplier<T> fx) {
            return new UnaryOperator<T>() {
                public T apply(T t) {
                    return fx.get();
                };
            };
        }

        // void->void
        private UnaryOperator<T> voidable(Runnable fx) {
            return new UnaryOperator<T>() {
                public T apply(T t) {
                    fx.run();
                    return t;
                };
            };
        }

    }

    /*
     * 流操作工具包
     */
    public static class StreamUtils<T> {
        private Stream<T> stream;
        final private Flow<T> flow;

        public StreamUtils(Flow<T> flow) {
            this.flow = flow;
        }

        public StreamUtils<T> batch(Stream<T> dataStream, String k) {
            dataStream.forEach(d -> flow.start(k, d));
            return this;
        }

        public StreamUtils<T> pipe(String k) {
            return pipe(stream, k);
        }

        public StreamUtils<T> pipe(Stream<T> dataStream, String k) {
            stream = dataStream.map(d -> {
                // 取消并发，以便于共同修改data
                flow.start(k, d, -1);
                return d;
            });
            return this;
        }

        public List<T> collect() {
            return stream.collect(Collectors.toList());
        }
    }

}